package org.cuasloi.mqtt.config;

import lombok.Data;
import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import static org.cuasloi.mqtt.constants.MqttConstants.*;


@Configuration
@ConfigurationProperties(prefix = "mqtt")
@Data
@Log4j2
public class MqttConfig {

    private String username;

    private String password;

    private String url;

    private String clientId;

    private String defaultTopic;

    @Autowired
    @Qualifier("input")
    private MessageHandler messageHandler;

    /**
     * Mqtt 连接配置
     */
    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        // 设置连接的用户名
        options.setUserName(username);
        // 设置连接的密码
        options.setPassword(password.toCharArray());
        // 设置连接的地址
        options.setServerURIs(new String[]{url});
        // 设置连接超时时间 单位为秒
        options.setConnectionTimeout(10);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(20);
        // 最大并发连接数
        options.setMaxInflight(10000);
        // 设置自动重连
        options.setAutomaticReconnect(true);
        // 清除无效会话
        options.setCleanSession(true);
        // 设置 "遗嘱" 消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的"遗嘱"消息。
        options.setWill("willTopic", WILL_DATA, 2, false);
        return options;
    }

    /**
     * Mqtt 客户端工厂
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions());
        return factory;
    }

    /**
     * 创建一个名为CHANNEL_OUT消息管道，用于发送消息
     */
    @Bean(name = CHANNEL_OUT)
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * 发送消息处理器
     * 消息默认处理配置，@ServiceActivator表明此方法会处理CHANNEL_OUT通道的消息
     */
    @Bean
    @ServiceActivator(inputChannel = CHANNEL_OUT)
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                clientId + "-publisher" + System.currentTimeMillis(),
                mqttClientFactory()
        );
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(defaultTopic);
        messageHandler.setDefaultQos(0);
        return messageHandler;
    }

    /**
     * 创建一个名为CHANNEL_IN消息管道， 用于接收消息
     */
    @Bean(name = CHANNEL_IN)
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * 配置订阅消息，订阅的主题
     */
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                clientId + "-subscriber" + System.currentTimeMillis(),
                mqttClientFactory(),
                "$queue/" + defaultTopic
        );
        adapter.setCompletionTimeout(3000);
        // 设置转换器，接收bytes
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(true);
        adapter.setConverter(converter);

        adapter.setQos(0);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    /**
     * 订阅消息处理， @ServiceActivator表明此方法会处理CHANNEL_IN通道的消息
     */
    @Bean
    @ServiceActivator(inputChannel = CHANNEL_IN)
    public MessageHandler handler() {
        return messageHandler;
    }
}
